#include "rpc_channel.h"

#include "rpc_controller.h"
#include "rpc_header.pb.h"

#include <arpa/inet.h>
#include <cerrno>
#include <netinet/in.h>
#include <string>
#include <sys/socket.h>
#include <unistd.h>
// #include "util.h"
// #include "rpc_example/friend.pb.h"
// class fixbug::GetFriendsListRequest;
/*
header_size + service_name method_name args_size + args
*/
// 所有通过stub代理对象调用的rpc方法，都会走到这里了，
// 统一通过rpcChannel来调用方法
// 统一做rpc方法调用的数据数据序列化和网络发送

void MprpcChannel::CallMethod(const google::protobuf::MethodDescriptor* method,
                              google::protobuf::RpcController*          controller,
                              const google::protobuf::Message*          request,
                              google::protobuf::Message*                response,
                              google::protobuf::Closure*                done) {
    // response 用户传入的，期望被填充
    // done  用户传入的，期望RPC完成后被调用
    std::lock_guard<std::mutex> lock(m_call_mutex); // 在方法开始时加锁

    std::cout << "MprpcChannel::CallMethod for " << method->full_name() << std::endl;
    MprpcController* mc = static_cast<MprpcController*>(controller); // 假设可以转换

    // 1. 确保连接 (错误处理时需要调用 done)
    if (m_clientFd == -1) {
        std::string errMsg;
        bool        rt = newConnect(m_ip.c_str(), m_port, &errMsg);
        if (!rt) {
            if (mc)
                mc->SetFailed("Connect failed: " + errMsg);
            if (done)
                done->Run(); // 关键：即使连接失败，也要执行回调通知上层
            return;
        }
    }

    // 2. 序列化请求
    const google::protobuf::ServiceDescriptor* sd           = method->service();
    std::string                                service_name = sd->name();     // service_name
    std::string                                method_name  = method->name(); // method_name

    // 获取参数的序列化字符串长度 args_size
    std::size_t args_size{};
    std::string args_str;
    if (!request->SerializeToString(&args_str)) {
        if (mc)
            mc->SetFailed("serialize request error!");
        if (done)
            done->Run();
        return;
    }
    args_size = args_str.size();

    RPC::RpcHeader rpcHeader; // 自定义的RPC头部
    rpcHeader.set_service_name(service_name);
    rpcHeader.set_method_name(method_name);
    rpcHeader.set_args_size(static_cast<google::protobuf::uint32>(args_size));
    std::string rpc_header_str;
    if (!rpcHeader.SerializeToString(&rpc_header_str)) {
        if (mc)
            mc->SetFailed("serialize rpc header error!");
        if (done)
            done->Run();
        return;
    }

    // 使用protobuf的CodedOutputStream来构建发送的数据流
    std::string send_rpc_str; // 用来存储最终发送的数据
    {
        // 创建一个StringOutputStream用于写入send_rpc_str
        google::protobuf::io::StringOutputStream string_output(&send_rpc_str);
        google::protobuf::io::CodedOutputStream  coded_output(&string_output);
        // 先写入header的长度（变长编码）

        coded_output.WriteVarint32(static_cast<uint32_t>(rpc_header_str.size()));
        // 然后写入rpc_header本身
        coded_output.WriteString(rpc_header_str);
    }
    // 最后，将请求参数附加到send_rpc_str后面
    send_rpc_str += args_str;

    // 3. 发送请求 (重试逻辑可以保留)
    // 注意：如果 send 失败多次最终放弃，也需要调用 done->Run()
    int send_attempts = 3; // 最多重试3次
    while (::send(m_clientFd, send_rpc_str.c_str(), send_rpc_str.size(), 0) == -1) {
        send_attempts--;
        char errtxt[512] = {0};
        sprintf(errtxt, "send error! errno:%d. Attempts left: %d", errno, send_attempts);
        std::cout << "Send error, attempting reconnect. IP: " << m_ip << " Port: " << m_port << std::endl;
        // 关闭当前连接
        close(m_clientFd);
        m_clientFd = -1;
        if (send_attempts <= 0) {
            if (mc)
                mc->SetFailed(std::string("Send failed after multiple retries: ") + errtxt);
            if (done)
                done->Run();
            return;
        }
        // 尝试重新连接
        std::string errMsg;
        if (!newConnect(m_ip.c_str(), m_port, &errMsg)) {
            if (mc)
                mc->SetFailed("Reconnect failed during send: " + errMsg);
            if (done)
                done->Run();
            return;
        }
    }
    // std::cout << "请求发送成功 (Method: " << method_name << ", User: "
    //           << (dynamic_cast<const fixbug::GetFriendsListRequest*>(request)
    //                   ? std::to_string(dynamic_cast<const fixbug::GetFriendsListRequest*>(request)->userid())
    //                   : "N/A")
    //           << ")" << std::endl;

    // 4. 接收和解析响应 (大改)
    //    需要一个循环来确保读取到完整的RPC帧，并正确解析。这里我们简化，但指出关键点。

    // ---- 正确的接收和解析逻辑 ----
    // A. 读取RPC响应帧的头部长度 (Varint32)
    uint32_t response_header_size = 0;
    char     header_len_buf[5]; // Varint32 最多5字节
    int      bytes_read_for_varint = 0;
    bool     varint_fully_read     = false;
    for (int i = 0; i < 5; ++i) {
        ssize_t nbytes = recv(m_clientFd, &header_len_buf[i], 1, 0);
        if (nbytes <= 0) { // 错误或连接关闭
            if (mc)
                mc->SetFailed(std::string("Recv error/closed while reading response header_len: ")
                              + (nbytes == 0 ? "closed" : strerror(errno)));
            close(m_clientFd);
            m_clientFd = -1;
            if (done)
                done->Run();
            return;
        }
        bytes_read_for_varint++;
        if ((static_cast<unsigned char>(header_len_buf[i]) & 0x80) == 0) {
            varint_fully_read = true;
            break;
        }
    }
    if (!varint_fully_read) {
        if (mc)
            mc->SetFailed("Malformed Varint for response header_len (too long).");
        close(m_clientFd);
        m_clientFd = -1; // 通常表示连接有问题
        if (done)
            done->Run();
        return;
    }
    { // 解析 Varint
        google::protobuf::io::ArrayInputStream arr_in(header_len_buf, bytes_read_for_varint);
        google::protobuf::io::CodedInputStream coded_in(&arr_in);
        if (!coded_in.ReadVarint32(&response_header_size)) {
            if (mc)
                mc->SetFailed("Failed to parse Varint32 for response header_size.");
            close(m_clientFd);
            m_clientFd = -1;
            if (done)
                done->Run();
            return;
        }
    }
    if (response_header_size == 0 || response_header_size > 65536) { // 合理性检查
        if (mc)
            mc->SetFailed("Invalid response_header_size: " + std::to_string(response_header_size));
        close(m_clientFd);
        m_clientFd = -1;
        if (done)
            done->Run();
        return;
    }

    // B. 读取RPC响应头部字符串
    std::string response_rpc_header_str;
    response_rpc_header_str.resize(response_header_size);
    ssize_t total_header_read = 0;
    while (total_header_read < response_header_size) {
        ssize_t nbytes =
            recv(m_clientFd, &response_rpc_header_str[total_header_read], response_header_size - total_header_read, 0);
        if (nbytes <= 0) {
            if (mc)
                mc->SetFailed(std::string("Recv error/closed while reading response RPC header: ")
                              + (nbytes == 0 ? "closed" : strerror(errno)));
            close(m_clientFd);
            m_clientFd = -1;
            if (done)
                done->Run();
            return;
        }
        total_header_read += nbytes;
    }
    RPC::RpcHeader response_rpc_header_pb; // 你定义的RPC头部 message
    if (!response_rpc_header_pb.ParseFromString(response_rpc_header_str)) {
        if (mc)
            mc->SetFailed("Failed to parse response RPC header string.");
        // 可能需要关闭连接，因为流可能已损坏
        if (done)
            done->Run();
        return;
    }
    uint32_t response_args_size = response_rpc_header_pb.args_size();

    // C. 读取RPC响应参数 (实际的 Protobuf Message)
    std::string response_args_str;
    response_args_str.resize(response_args_size);
    ssize_t total_args_read = 0;
    while (total_args_read < response_args_size) {
        ssize_t nbytes = recv(m_clientFd, &response_args_str[total_args_read], response_args_size - total_args_read, 0);
        if (nbytes <= 0) {
            if (mc)
                mc->SetFailed(std::string("Recv error/closed while reading response args: ")
                              + (nbytes == 0 ? "closed" : strerror(errno)));
            close(m_clientFd);
            m_clientFd = -1;
            if (done)
                done->Run();
            return;
        }
        total_args_read += nbytes;
    }
    // std::cout << "请求响应成功 (Method: " << method_name << ", User: "
    //           << (dynamic_cast<const fixbug::GetFriendsListRequest*>(request)
    //                   ? std::to_string(dynamic_cast<const fixbug::GetFriendsListRequest*>(request)->userid())
    //                   : "N/A")
    //           << ")" << std::endl;

    // D. 将响应参数反序列化到用户传入的 response 对象
    if (!response->ParseFromString(response_args_str)) {
        char errtxt[1050] = {0};
        // 注意：response_args_str 可能包含二进制数据，直接 sprintf 可能会截断或出错
        sprintf(errtxt, "Parse response_args_str error! (Size: %u)", response_args_size);
        if (mc)
            mc->SetFailed(errtxt);
        if (done)
            done->Run();
        return;
    } else {
        // if (mc) mc->SetFailed(false); // 标记成功
        // if (mc) mc->SetFailed(""); //标记成功
    }

    // 5. 调用 done 回调 (无论成功或失败)
    if (done) {
        done->Run();
    }
}

bool MprpcChannel::newConnect(const char* ip, uint16_t port, string* errMsg) {
    std::cout << "MprpcChannel::newConnect is called" << std::endl;
    int clientfd = socket(AF_INET, SOCK_STREAM, 0);
    if (-1 == clientfd) {
        char errtxt[512] = {0};
        sprintf(errtxt, "create socket error! errno:%d", errno);
        m_clientFd = -1;
        *errMsg    = errtxt;
        return false;
    }

    struct sockaddr_in server_addr;
    server_addr.sin_family      = AF_INET;
    server_addr.sin_port        = htons(port);
    server_addr.sin_addr.s_addr = inet_addr(ip);
    // 连接rpc服务节点
    if (-1 == connect(clientfd, reinterpret_cast<struct sockaddr*>(&server_addr), sizeof(server_addr))) {
        close(clientfd);
        char errtxt[512] = {0};
        sprintf(errtxt, "connect fail! errno:%d", errno);
        m_clientFd = -1;
        *errMsg    = errtxt;
        return false;
    }
    m_clientFd = clientfd;
    return true;
}

MprpcChannel::MprpcChannel(string ip, short port, bool connectNow)
    : m_clientFd(-1)
    , m_ip(ip)
    , m_port(port) {
    // 使用tcp编程，完成rpc方法的远程调用，使用的是短连接，因此每次都要重新连接上去，待改成长连接。
    // 没有连接或者连接已经断开，那么就要重新连接呢,会一直不断地重试
    // 读取配置文件rpcserver的信息
    // std::string ip = MprpcApplication::GetInstance().GetConfig().Load("rpcserverip");
    // uint16_t port = atoi(MprpcApplication::GetInstance().GetConfig().Load("rpcserverport").c_str());
    // rpc调用方想调用service_name的method_name服务，需要查询zk上该服务所在的host信息
    //  /UserServiceRpc/Login

    if (!connectNow) {
        return;
    } //可以允许延迟连接
    std::string errMsg;
    auto        rt       = newConnect(ip.c_str(), port, &errMsg);
    int         tryCount = 3;
    while (!rt && tryCount--) {
        std::cout << errMsg << std::endl;
        rt = newConnect(ip.c_str(), port, &errMsg);
    }
}

MprpcChannel::~MprpcChannel() {
    if (m_clientFd != -1) {
        std::cout << "MprpcChannel destructor: Closing fd " << m_clientFd << " for IP " << m_ip << ":" << m_port
                  << std::endl;
        close(m_clientFd);
        m_clientFd = -1;
    }
}